package band.wukong.practice.java.core.topic.concurrency.threadpool;

import java.util.concurrent.*;

/**
 * CompletionService版的计算器。
 * <br>
 * 在ConcurrentCalculator1中，getResult()方法的实现过程中迭代了FutureTask的数组。如果任务还没有完成则当前线程会阻塞。
 * <br>
 * 如果我们希望任意子任务完成后就把其结果加到result中，而不用依次等待每个任务完成，可以使CompletionService。
 * 生产者submit()执行的任务。使用者take()已完成的任务，并按照完成这些任务的顺序处理它们的结果 。
 *
 * @author wukong(wukonggg@139.com)
 */
public class ConcurrentCalculator2 {

    private int cpuCoreNumber;
    private ExecutorService exec;
    private CompletionService<Long> completionService;

    public ConcurrentCalculator2() {
        cpuCoreNumber = Runtime.getRuntime().availableProcessors();
        exec = Executors.newFixedThreadPool(cpuCoreNumber);
        completionService = new ExecutorCompletionService<>(exec);
    }

    public static void main(String[] args) {
        int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 10, 11 };
        ConcurrentCalculator2 calc = new ConcurrentCalculator2();
        Long sum = calc.sum(numbers);
        System.out.println(sum);
        calc.close();
    }

    private Long sum(final int[] numbers) {
        // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor
        for (int i = 0; i < cpuCoreNumber; i++) {
            int increment = numbers.length / cpuCoreNumber + 1;
            int start = increment * i;
            int end = increment * i + increment;
            if (end > numbers.length) {
                end = numbers.length;
            }
            SumCalculator subCalc = new SumCalculator(numbers, start, end);
            if (!exec.isShutdown()) {
                completionService.submit(subCalc);
            }
        }
        System.out.println("cpuCoreNumber = " + cpuCoreNumber);
        return getResult();
    }

    /**
     * 迭代每个子任务，获得部分和，相加返回
     */
    private Long getResult() {
        Long result = 0L;
        for (int i = 0; i < cpuCoreNumber; i++) {
            System.out.println("i = " + i);
            try {
                Long subSum = completionService.take().get();
                result += subSum;
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    private void close() {
        exec.shutdown();
    }

    static class SumCalculator implements Callable<Long> {
        private int[] numbers;
        private int start;
        private int end;

        SumCalculator(final int[] numbers, int start, int end) {
            this.numbers = numbers;
            this.start = start;
            this.end = end;
        }

        public Long call() throws Exception {
            Long sum = 0L;
            for (int i = start; i < end; i++) {
                sum += numbers[i];
            }
            return sum;
        }
    }

}
